package com.arch.flume.interceptor;

import com.alibaba.fastjson2.JSON;
import com.google.common.base.Strings;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 物联网的部分数据会保存在边缘设备上，直到下次开机进行上传，因此在用flume进行数据搜集的时候会存在补发的问题
 * 落分区应该按照事件时间而不是flume主机的时间
 * 事件时间拦截器则是为了应对以上场景
 *
 * https://cloud.tencent.com/developer/article/1695238
 *
 * @author Eights
 */
public class EventTimeInterceptor implements Interceptor {

    private static FastDateFormat dateFormat = FastDateFormat.getInstance("yyyy-MM-dd");

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        //获取header
        Map<String, String> headers = event.getHeaders();

        //获取body
        String eventBody = new String(event.getBody(), StandardCharsets.UTF_8);

        String[] bodyArr = eventBody.split("\\s+");

        try {
            String jsonStr = bodyArr[6];

            //数据为空，返回null
            if (Strings.isNullOrEmpty(jsonStr)) {
                return null;
            }

            long ts = Long.parseLong(JSON.parseObject(jsonStr).getJSONObject("app_active").getString("time"));
            //打上事件日期
            String eventDate = dateFormat.format(ts);
            //header中添加event date
            headers.put("eventDate", eventDate);
            event.setHeaders(headers);
        } catch (Exception e) {
            //脏数据,需要sink到一个目录进行核查
            headers.put("eventDate", "unknow");
            event.setHeaders(headers);
        }

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {

        return list.stream().map(this::intercept)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());

    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new EventTimeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
